import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.*;

public class T6RPC {
    static final String QUEUE_NAME = "rpc_queue";
    static final ExecutorService es = Executors.newFixedThreadPool(2);

    public static void main(String[] args) {
        es.submit(T6RPC::rpcServer);
        es.submit(T6RPC::rpcClient);
    }

    static int fib(int n) {
        if (n == 0) return 0;
        if (n == 1) return 1;
        return fib(n - 1) + fib(n - 2);
    }

    static void rpcServer() {
        try {
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
            // 创建队列
            // 参数 1：队列名称
            // 参数 2：是否定义持久化队列
            // 参数 3：是否独占本次连接
            // 参数 4：是否在不使用的时候自动删除队列
            // 参数 5：队列其它参数
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            channel.queuePurge(QUEUE_NAME); // 清空指定队列中的消息
            channel.basicQos(1); // 公平调度，一次只接受一条未确认的消息
            System.out.println(Thread.currentThread().getName() + " [x] Awaiting RPC requests");
            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                AMQP.BasicProperties replyProps = new AMQP.BasicProperties
                    .Builder()
                    .correlationId(delivery.getProperties().getCorrelationId())
                    .build();
                String response = "";
                try {
                    String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                    System.out.println(Thread.currentThread().getName() + " [.] fib(" + message + ")");
                    response += fib(Integer.parseInt(message));
                } catch (RuntimeException e) {
                    System.out.println(Thread.currentThread().getName() + " [.] " + e);
                } finally {
                    // 发送消息
                    // 参数 1：交换机名称，如果没有指定则使用默认 Default Exchage
                    // 参数 2：路由 key，简单模式可以传递队列名称
                    // 参数 3：消息属性
                    // 参数 4：消息内容
                    channel.basicPublish("", delivery.getProperties().getReplyTo(), replyProps, response.getBytes("UTF-8"));
                    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); // 确认消息
                }
            };
            // 订阅消息
            // 爹数 1：队列名称
            // 参数 2：是否自动确认。设置为 true 表示消息按收到后自动向 mq 回复接收到了，mq 接收到回复后会删除消息：设置为 false 则需要手动确认
            channel.basicConsume(QUEUE_NAME, false, deliverCallback, (consumerTag -> {
            }));
        } catch (IOException | TimeoutException e) {
            e.printStackTrace();
        }
    }

    public static void rpcClient() {
        try (Connection connection = ConnectionUtil.getConnection();
             Channel channel = connection.createChannel()) {
            for (int i = 0; i < 32; i++) {
                // 创建一个非持久的、独占的、具有生成名称的自动删除队列
                String replyQueueName = channel.queueDeclare().getQueue();
                String corrId = UUID.randomUUID().toString();
                AMQP.BasicProperties props = new AMQP.BasicProperties
                    .Builder()
                    .correlationId(corrId)
                    .replyTo(replyQueueName)
                    .build();
                String message = Integer.toString(i);
                System.out.println(Thread.currentThread().getName() + " [x] Requesting fib(" + message + ")");
                // 发送消息
                // 参数 1：交换机名称，如果没有指定则使用默认 Default Exchage
                // 参数 2：路由 key，简单模式可以传递队列名称
                // 参数 3：消息属性
                // 参数 4：消息内容
                channel.basicPublish("", QUEUE_NAME, props, message.getBytes(StandardCharsets.UTF_8));
                CompletableFuture<String> future = new CompletableFuture<>();
                // 订阅消息
                // 爹数 1：队列名称
                // 参数 2：是否自动确认。设置为 true 表示消息按收到后自动向 mq 回复接收到了，mq 接收到回复后会删除消息：设置为 false 则需要手动确认
                String ctag = channel.basicConsume(replyQueueName, true, (consumerTag, delivery) -> {
                    if (delivery.getProperties().getCorrelationId().equals(corrId)) {
                        future.complete(new String(delivery.getBody(), StandardCharsets.UTF_8));
                    }
                }, consumerTag -> {
                });
                String response = future.get();
                channel.basicCancel(ctag); // 取消订阅
                System.out.println(Thread.currentThread().getName() + " [.] Got '" + response + "'");
            }
        } catch (IOException | TimeoutException | ExecutionException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}